package cn.doitedu.rtdw.feature_etl;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/19
 * @Desc: 学大数据，上多易教育
 *   广告点击率预估特征数据加工
 **/
public class AdTransformRationPredictFeatures {
    public static void main(String[] args) {


        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "     event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );

        // 利用flink-cep在行为日志流中寻找“曝光”后面有“点击”的模式
        tenv.executeSql(
                        "create temporary view show_click as                       \n "+
                        "with ad_events AS (                                        \n "+
                        "     SELECT                                                \n "+
                        "       user_id,                                            \n "+
                        " 		event_id,                                           \n "+
                        " 		action_time,                                        \n "+
                        " 		properties['ad_id'] as ad_id,                       \n "+
                        " 		properties['ad_tracking_id'] as ad_tracking_id,     \n "+
                        " 		rt                                                  \n "+
                        "     from dwd_events_kafka                                 \n "+
                        "     where event_id in ('ad_show','ad_click','ad_transform')   \n "+
                        " )                                                         \n "+
                        "                                                           \n "+
                        "                                                           \n "+
                        " SELECT                                                    \n "+
                        "      *,proctime() as pt                                   \n "+
                        " from ad_events                                            \n "+
                        " match_recognize(                                          \n "+
                        "     partition by user_id,ad_tracking_id                   \n "+
                        " 	order by rt                                             \n "+
                        " 	measures                                                \n "+
                        " 	    A.user_id as uid,                                   \n "+
                        " 		A.ad_id as adid,                                    \n "+
                        " 		A.ad_tracking_id as tkid,                           \n "+
                        " 		A.event_id as show_event,                           \n "+
                        " 		A.action_time as show_time,                         \n "+
                        " 		first(B.event_id,0) as click_event,                 \n "+
                        " 		first(B.action_time,0) as click_time,               \n "+
                        " 		C.event_id as transform_event,                      \n "+
                        " 		C.action_time as transform_time                    \n "+
                        " 	one row per match                                       \n "+
                        " 	after match skip past last row                          \n "+
                        " 	pattern (A B+ C)                                        \n "+
                        " 	define                                                  \n "+
                        " 	   A AS A.event_id = 'ad_show',                         \n "+
                        " 	   B AS B.event_id = 'ad_click',                        \n "+
                        " 	   C AS C.event_id = 'ad_transform'                     \n "+
                        " )                                                         \n "
        );




        // 创建hbase中的请求特征日志数据映射表
        tenv.executeSql(
                " create table ad_request_features_log_hbase (     "+
                        " 	ad_tracking_id string,                 "+
                        "   f  row<log_data  string>                    "+
                        " ) WITH (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'ad_request_features_log',   "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );


        // 将上面的流内关联结果，  用lookup关联这张 请求特征日志表
        tenv.executeSql(
                        " select                                        "+
                        "     a.uid as user_id,                         "+
                        "     a.tkid as ad_tracking_id,                 "+
                        "     a.show_event as show_event_id,            "+
                        "     a.show_time,                              "+
                        "     a.click_event as click_event_id,          "+
                        "     a.click_time,                             "+
                        "     a.transform_event as transform_event_id,  "+
                        "     a.transform_time,                         "+
                        "     b.f.log_data as request_log_data          "+
                        " from show_click a                             "+
                        " left join ad_request_features_log_hbase       "+
                        " FOR SYSTEM_TIME AS OF a.pt  as b              "+
                        " on a.tkid = b.ad_tracking_id                  "
        ).print();






    }


}
